package io.kiki.sba.registry.server.clientInterface.remoting.handler;

import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.dataserver.DatumSummary;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherRequest;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffPublisherResult;
import io.kiki.sba.registry.common.model.slot.DataSlotDiffUtils;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptorManager;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.bootstrap.ExecutorManager;
import io.kiki.sba.registry.server.clientInterface.slot.SlotsCache;
import io.kiki.sba.registry.server.clientInterface.store.Publishers;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;

public class DataSlotDiffPublisherRequestHandler extends AbstractServerHandler<DataSlotDiffPublisherRequest> {

    private static final Logger logger = LoggerFactory.getLogger(DataSlotDiffPublisherRequestHandler.class);

    @Autowired
    Config config;

    @Autowired
    ExecutorManager executorManager;

    @Autowired
    Publishers publishers;

    @Autowired
    SlotsCache slotsCache;

    @Resource
    SyncSlotAcceptorManager syncSlotAcceptAllManager;

    @Override
    public void checkParam(DataSlotDiffPublisherRequest request) {
        StringUtil.checkNonNegative(request.getSlotId(), "request.slotId");
        StringUtil.nullValue(request.getDatumSummaries(), "request.datumSummaries");
    }

    @Override
    public Object doHandle(Channel channel, DataSlotDiffPublisherRequest request) {
        try {
            final int slotId = request.getSlotId();
            DataSlotDiffPublisherResult result = calcDiffResult(request.getLocalDataCenter(), slotId, request.getDatumSummaries(), publishers.getDataInfoIdPublishers(slotId));
            result.setSlotsEpoch(slotsCache.getEpoch(request.getLocalDataCenter()));
            result.setClientInterfaceId(ServerEnv.PROCESS_ID);
            return new GenericResponse().fillSucceed(result);
        } catch (Throwable e) {
            String msg = StringFormatter.format("DiffSyncPublisher request error for slot {}", request.getSlotId());
            logger.error(msg, e);
            return new GenericResponse().fillFailed(msg);
        }
    }

    private DataSlotDiffPublisherResult calcDiffResult(String requestDataCenter, int targetSlot, List<DatumSummary> datumSummaries, Map<String, Map<String, Publisher>> existingPublishers) {
        DataSlotDiffPublisherResult result = DataSlotDiffUtils.diffPublishersResult(existingPublishers, datumSummaries, config.getSlotSyncPublisherMaxNum(), syncSlotAcceptAllManager);
        DataSlotDiffUtils.logDiffResult(requestDataCenter, result, targetSlot);
        return result;
    }

    public Object buildFailedResponse(String msg) {
        return new GenericResponse().fillFailed(msg);
    }

    @Override
    protected Node.NodeType getConnectNodeType() {
        return Node.NodeType.data_store;
    }

    @Override
    public Executor getExecutor() {
        return executorManager.getDataSlotSyncRequestExecutor();
    }

    @Override
    public Class interest() {
        return DataSlotDiffPublisherRequest.class;
    }
}
